package org.zjt.chapter1.rxnetty;

import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.logging.LogLevel;
import io.reactivex.netty.RxNetty;
import io.reactivex.netty.protocol.http.client.HttpClient;
import io.reactivex.netty.protocol.http.client.HttpClientRequest;
import io.reactivex.netty.protocol.http.client.HttpClientResponse;
import io.reactivex.netty.protocol.http.server.*;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import rx.Observable;
import io.netty.buffer.ByteBuf;
import rx.functions.Func1;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

import static rx.Observable.*;

/**
 * Description:
 *
 * @author juntao.zhang
 * Date: 2018-07-18 下午8:00
 * @see
 */
@Slf4j
public class RxNettyTest {


    @Test
    public void server()throws Exception {

        RequestRouterHandler requestRouterHandler = new RequestRouterHandler();
        HttpServer<ByteBuf, ByteBuf> server = HttpServer.newServer(9090).start(requestRouterHandler);

        requestRouterHandler.addRequestHandler("/home",(req,res) -> res.writeString(just("hello, welcome to home !")));

        /*Wait for shutdown if not called from the client (passed an arg)*/
        server.awaitShutdown();

    }


    public static Func1<ByteBuf,String> getByteBuf2StringFunc(){
        return byteBuf -> {
            byte[] bytes = new byte[byteBuf.readableBytes()];
            byteBuf.readBytes(bytes);
            return new String(bytes);
        };
    }



    @Test
    public void interceptClient() {
        HttpClientRequest<ByteBuf, ByteBuf> getResult = HttpClient.newClient("localhost", 9090)
                .enableWireLogging("hello-client", LogLevel.INFO)
                .intercept()
                .next(provider -> ((version, method, uri) -> provider.createRequest(version, method, uri).addHeader("zhang", "ssss")))
                .finish()
                .createGet("/hello");


        getResult.map(resp -> resp.getContent().map(getByteBuf2StringFunc())).toBlocking()
                .subscribe(content ->  content.subscribe(str -> log.info("content :{}",str)));
    }


    @Test
    public void client()throws Exception {

        HttpClient.newClient("localhost",9090)
                .enableWireLogging("hello-client", LogLevel.INFO)
                .createGet("/hello")
                .doOnCompleted(() -> log.info("执行结束"))
                .doOnNext(resp -> log.info(resp.toString()))
                .map(resp -> resp.getContent().map(getByteBuf2StringFunc()))
                .toBlocking()
                .forEach(ss ->  ss.subscribe(str -> log.info("content :{}",str)));


        TimeUnit.SECONDS.sleep(8);
    }


    public static class RequestRouterHandler implements RequestHandler<ByteBuf, ByteBuf> {

        private ConcurrentHashMap<String,RequestHandler> routerMap = new ConcurrentHashMap<>();

        public void addRequestHandler(String path ,RequestHandler requestHandler){
            Objects.requireNonNull(requestHandler,"requestHandler can not is null !");
            Objects.requireNonNull(path,"path can not is null !");
            routerMap.put(path,requestHandler);
        }


        @Override
        public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) {
            String uri = request.getUri();
            return routerMap.getOrDefault(uri, (request1, response1) -> response1.writeString(just("404：未找到请求的URL"))).handle(request,response);
        }
    }
}
